package com.edu.yx.app.App_04_DwsApp.Dws_05_Exam;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp;
import com.edu.yx.bean.ExamQuestionBean;
import com.edu.yx.bean.ExamQuestionResultBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;

import static com.edu.yx.common.Constant.*;

public class Dws_04_Exam_Question_Count extends BaseApp {

    public static void main(String[] args) {
        new Dws_04_Exam_Question_Count().init(
                4054,
                2,
                "Dws_04_Exam_Question_Count",
                TOPIC_DWD_EXAM_PAPER_QUESTION_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.封装成pojo
        SingleOutputStreamOperator<ExamQuestionBean> parseToPojoStream = parseToPojo(stream);

//        parseToPojoStream.print();

        //2.开窗聚合
        SingleOutputStreamOperator<ExamQuestionBean> windowAndAggStream = windowAndAgg(parseToPojoStream);
        
        //3.根据需求进行计算并且二次封装pojo
        SingleOutputStreamOperator<ExamQuestionResultBean> toPojoAgainStream = parseToPojoAgain(windowAndAggStream);

        //4.写入doris
        writeDoris(toPojoAgainStream);

    }
    private void writeDoris(SingleOutputStreamOperator<ExamQuestionResultBean> resultStream) {
        resultStream
                .map(bean ->{
                    //配置json可以将驼峰转换成带下划线，然后方便导入数据到doris
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);
//                    System.out.println(json);
                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_exam_question_count_window"));

    }

    private SingleOutputStreamOperator<ExamQuestionResultBean> parseToPojoAgain(SingleOutputStreamOperator<ExamQuestionBean> windowAndAggStream) {

        return windowAndAggStream
                .map(bean -> {

                    //分子
                    Long correctQuestionCt = bean.getCorrectQuestionCt();
                    Long correctQuestionUserIdCt = bean.getCorrectQuestionUserIdCt();

                    //分母
                    Long questionCt = bean.getQuestionCt();
                    Long questionUserIdCt = bean.getQuestionUserIdCt();

                    //答题正确率
                    Double rightRate = (correctQuestionCt * 100.0) / questionCt / 100.0 ;
                    Double userRightRate = correctQuestionUserIdCt * 100.0 / questionUserIdCt / 100.0;

                    return new ExamQuestionResultBean(
                            bean.getStt(),
                            bean.getEdt(),
                            bean.getId(),
                            bean.getCurDate(),
                            correctQuestionCt,
                            questionCt,
                            rightRate,
                            correctQuestionUserIdCt,
                            questionUserIdCt,
                            userRightRate,
                            bean.getTs()
                    );

                });

    }

    private SingleOutputStreamOperator<ExamQuestionBean> windowAndAgg(SingleOutputStreamOperator<ExamQuestionBean> parseToPojoStream) {

        SingleOutputStreamOperator<ExamQuestionBean> reduce = parseToPojoStream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<ExamQuestionBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean -> bean.getId())
                .window(TumblingEventTimeWindows.of(Time.hours(18)))
                .reduce(new ReduceFunction<ExamQuestionBean>() {

                            @Override
                            public ExamQuestionBean reduce(ExamQuestionBean value1,
                                                           ExamQuestionBean value2) throws Exception {

                                value1.setQuestionCt(value1.getQuestionCt() + value2.getQuestionCt());
                                value1.setCorrectQuestionCt(value1.getCorrectQuestionCt() + value2.getCorrectQuestionCt());

                                value1.getCorrectQuestionUserIdSet().addAll(value2.getCorrectQuestionUserIdSet());
                                value1.getQuestionUserIdSet().addAll(value2.getQuestionUserIdSet());


                                return value1;
                            }
                        },
                        new ProcessWindowFunction<ExamQuestionBean, ExamQuestionBean, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context ctx,
                                                Iterable<ExamQuestionBean> elements,
                                                Collector<ExamQuestionBean> out) throws Exception {

                                ExamQuestionBean bean = elements.iterator().next();

                                bean.setStt(TempUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));


                                bean.setQuestionUserIdCt((long) bean.getQuestionUserIdSet().size());

                                //正确的要处理掉前面补充进去的null，留下user_id
                                HashSet<String> correctQuestionUserIdSet = bean.getCorrectQuestionUserIdSet();
                                correctQuestionUserIdSet.removeAll(Collections.singleton("null"));
                                bean.setCorrectQuestionUserIdCt((long) correctQuestionUserIdSet.size());


                                out.collect(bean);
                            }
                        }

                );

        return reduce;

    }

    private SingleOutputStreamOperator<ExamQuestionBean> parseToPojo(SingleOutputStreamOperator<String> stream) {

        //流的数据为JSONObject类型，先将数据封装成pojo，方便后续计算
        return stream
                .map(strjson->{

                    //转成json对象
                    JSONObject jsonObject = JSON.parseObject(strjson);

//                    new HashSet<>(Collections.singleton(value.getString("order_id"))))

                    String userId = jsonObject.getString("user_id");
                    String questionId = jsonObject.getString("id");
                    String isCorrect = jsonObject.getString("is_correct");
                    Long ts = jsonObject.getLong("ts") * 1000L ;

                    Long questionCt = 1L;
                    Long correctQuestionCt = 0L;

                    if ("1".equals(isCorrect)){
                        correctQuestionCt =1L ;
                    }

                    ExamQuestionBean examQuestionBean = new ExamQuestionBean(
                            "",
                            "",
                            questionId,
                            "",
                            correctQuestionCt,
                            questionCt,
                            "1".equals(isCorrect)== true? new HashSet<>(Collections.singleton(jsonObject.getString("user_id"))) : new HashSet<>(Collections.singleton("null")) ,
                            0L,
                            new HashSet<>(Collections.singleton(jsonObject.getString("user_id"))),
                            0L,
                            ts
                    );

                    return examQuestionBean;

                });


    }

}
